package xj.toolkit.access.dispatcher;

import java.util.Collection;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.RandomUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;

import xj.toolkit.netty.transport.TransportUtils;
import xj.toolkit.access.AccessHeader;
import xj.toolkit.access.connector.AccessTcpClientEndpoint;
import xj.toolkit.access.msg.AccessAckMessage;
import xj.toolkit.access.msg.AccessMessage;
import xj.toolkit.base.Sender;

/**
 * 从前端来的消息，需要分发到不同的目标channel。
 * 
 * @author <a href="mailto:cxj2000@gmail.com">xiaojian.cao</a>
 *
 */
public class DefaultMessageDispatcher implements MessageDispatcher, Sender<AccessMessage> {

	private static final Logger logger = LoggerFactory.getLogger(DefaultMessageDispatcher.class);

	private Map<ModuleInstancePair, Sender<AccessMessage>> moduleIdInstanceIdChannelMap = Maps.newConcurrentMap();

	private Multimap<Short, Short> moduleIdInstanceIdMap = ArrayListMultimap.create();
	
	public Map<Short, Collection<Short>> getSnapshotOfmoduleIdInstanceIdMap() {
		Map<Short, Collection<Short>> snapshot = Maps.newHashMap();
		for (Short moduleId : moduleIdInstanceIdMap.keySet()) {
			snapshot.put(moduleId, moduleIdInstanceIdMap.get(moduleId));
		}
		
		return snapshot;
	}

	/**
	 * 
	 * @param accessMessage
	 */
	public void onBusiModuleResp(AccessMessage accessMessage) {
		Preconditions.checkArgument(accessMessage != null);

		// -- 只响应AccessAckMessage消息。
		if (accessMessage instanceof AccessAckMessage) {
			Sender<AccessMessage> sender = TransportUtils.getSenderOf(accessMessage);
			if (sender == null) {
				logger.error("Cannot found sender from accessmessage[{}].", accessMessage);
				return;
			}

			AccessHeader header = accessMessage.getAccessHeader();
			if (header == null) {
				logger.error("Miss accessHeader accessmessage[{}].", accessMessage);
				return;
			}

			ModuleInstancePair pair = new ModuleInstancePair();
			pair.setModuleId(header.getSourceModuleId());
			pair.setInstanceId(header.getSourceInstanceId());
			moduleIdInstanceIdChannelMap.put(pair, sender);

			moduleIdInstanceIdMap.put(header.getSourceModuleId(), header.getSourceInstanceId());

			logger.debug("module[{}], instance[{}] --> Sender[{}].", header.getSourceModuleId(),
					header.getSourceInstanceId(), sender);
			return;
		}

		// -- 其他回应消息。
		
	}

	/**
	 * 当连接断开的时候。
	 * 
	 * @param sender
	 */
	public void onChannelDisconnect(Sender<AccessMessage> sender) {
		AccessTcpClientEndpoint clientEndpoint = (AccessTcpClientEndpoint) sender;
		if (clientEndpoint == null) {
			//TODO
		}
		
		short busiModuleId = clientEndpoint.getBusiModuleId();
		short busiInstanceId = clientEndpoint.getBusiInstanceId();
		
		if (busiInstanceId == 0x0 && busiModuleId == 0x0) {
			logger.error("Invalid endpoint[{}].", clientEndpoint);
			return;
		}
		
		ModuleInstancePair pair = new ModuleInstancePair();
		pair.setModuleId(clientEndpoint.getBusiModuleId());
		pair.setInstanceId(clientEndpoint.getBusiInstanceId());
		moduleIdInstanceIdChannelMap.remove(pair);
		
		logger.info("Remove instance[{}].", pair);
		
		moduleIdInstanceIdMap.remove(busiModuleId, busiInstanceId);
	}

	/**
	 * 发送accessMessage。 如果指明了instaceId，则检查instanceId是否连接。
	 * 如果没有指明instanceId，则随机指定一个。
	 * 
	 * @param target
	 */
	@Override
	public void send(AccessMessage target) {
		Preconditions.checkArgument(target != null);
		AccessHeader header = target.getAccessHeader();
		if (header == null) {
			logger.error("Miss accessHeader accessmessage[{}].", target);
			return;
		}

		// -- 已经指明了实例id。
		if (header.getTargetInstanceId() == 0) {
			List<Short> instanceIds = Lists.newArrayList(moduleIdInstanceIdMap.get(header.getTargetModuleId()));
			if (instanceIds == null || instanceIds.size() == 0) {
				logger.error("Cannot found any sender with module[{}]/instance[{}].", header.getTargetModuleId(),
						header.getTargetInstanceId());
				return;
			}

			Short instanceId = instanceIds.get(RandomUtils.nextInt(0, instanceIds.size()));
			header.setTargetInstanceId(instanceId);
		}

		Sender<AccessMessage> sender = moduleIdInstanceIdChannelMap
				.get(new ModuleInstancePair(header.getTargetModuleId(), header.getTargetInstanceId()));
		if (sender == null) {
			logger.error("Cannot found sender with module[{}]/instance[{}].", header.getTargetModuleId(),
					header.getTargetInstanceId());
			return;
		}

		sender.send(target);

	}

	@Override
	public void send(Object req, AccessMessage response) {
		send(response);
	}

	@Override
	public void dispatch(AccessMessage msg) {
		Preconditions.checkArgument(msg != null);
		this.send(msg);
	}

}
